package com.lesserpand.rabbitmq.rabbitmq.config;

import com.lesserpand.rabbitmq.rabbitmq.callback.MsgSendConfirmCallBack;
import com.lesserpand.rabbitmq.rabbitmq.callback.MsgSendReturnCallBack;
import com.lesserpand.rabbitmq.rabbitmq.constant.MqEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConversionException;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitMq配置
 *
 * @author pand
 */
@Configuration
public class RabbitMqConfig {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMqConfig.class);

    @Autowired
    private QueueConfig queueConfig;
    @Autowired
    private ExchangeConfig exchangeConfig;

    /**
     * 连接工厂
     */
    @Autowired
    private ConnectionFactory connectionFactory;


    /**
     * 消息中心实际消息交换器与队列绑定，第一次
     *
     * @param directExchange 消息中心交换配置
     * @param messageQueue   消息中心队列
     * @return
     */
    @Bean
    public Binding messageBinding(DirectExchange directExchange, Queue messageQueue) {
        return BindingBuilder.bind(messageQueue).to(directExchange).with(MqEnum.MESSAGE_QUEUE_KEY.getValue());
    }

    /**
     * 消息中心实际消息交换器与队列绑定，第二次
     *
     * @param directExchange2 消息中心交换配置
     * @param messageQueue2   消息中心队列
     * @return
     */
    @Bean
    public Binding messageBinding2(DirectExchange directExchange2, Queue messageQueue2) {
        return BindingBuilder.bind(messageQueue2).to(directExchange2).with(MqEnum.MESSAGE_QUEUE2_KEY.getValue());
    }

    /**
     * 延迟队列1与私信队列通过MESSAGE_TTL_QUEUE_KEY绑定
     *
     * @param messageTtlQueue
     * @param ttlDirectExchange
     * @return
     */
    @Bean
    public Binding messageTtlBinding(DirectExchange ttlDirectExchange, Queue messageTtlQueue) {
        return BindingBuilder.bind(messageTtlQueue).to(ttlDirectExchange).with(MqEnum.MESSAGE_TTL_QUEUE_KEY.getValue());
    }


    /**
     * 延迟队列2与私信队列通过MESSAGE_TTL_QUEUE2_KEY绑定
     *
     * @param ttlDirectExchange
     * @param messageTtlQueue2
     * @return
     */
    @Bean
    public Binding messageTtlBinding2(DirectExchange ttlDirectExchange, Queue messageTtlQueue2) {
        return BindingBuilder.bind(messageTtlQueue2).to(ttlDirectExchange).with(MqEnum.MESSAGE_TTL_QUEUE2_KEY.getValue());
    }


    /**
     * 将消息队列1和直连交换机进行绑定
     */
    @Bean
    public Binding binding_one() {
        return BindingBuilder.bind(queueConfig.firstQueue()).to(exchangeConfig.directExchange()).with(MqEnum.ROUTINGKEY1.getValue());
    }

    /**
     * 将消息队列2和交换机进行绑定
     */
    @Bean
    public Binding binding_two() {
        return BindingBuilder.bind(queueConfig.secondQueue()).to(exchangeConfig.directExchange()).with(MqEnum.ROUTINGKEY2.getValue());
    }

    /**
     * 将消息队列2和交换机进行绑定
     */
    @Bean
    public Binding skill_binding() {
        return BindingBuilder.bind(queueConfig.skillMessageQueue()).to(exchangeConfig.skillExchange()).with(MqEnum.SKILL_MESSAGE_QUEUE_KEY.getValue());
    }

    /**
     * queue listener  观察 监听模式，自己实现监听，当有消息时，调用该回调处理消息。如果不实现该方法，使用默认的。这里注释掉，使用默认消息处理。
     * 当有消息到达时会通知监听在对应的队列上的监听对象
     *
     * @return
     */
    // @Bean("messageListenerContainer")
    public SimpleMessageListenerContainer messageListenerContainer() {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.addQueues(queueConfig.firstQueue());
        simpleMessageListenerContainer.setExposeListenerChannel(true);
        //设置最大的并发的消费者数量
        simpleMessageListenerContainer.setMaxConcurrentConsumers(5);
        //最小的并发消费者的数量
        simpleMessageListenerContainer.setConcurrentConsumers(1);
        //设置确认模式手工确认
        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 消息处理回调，使用自己实现的监听器，这个监听必须实现。
        simpleMessageListenerContainer.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
            channel.basicQos(1);
            byte[] body = message.getBody();
            LOGGER.info("接收处理队列当中的消息:" + new String(body));
            /**为了保证永远不会丢失消息，RabbitMQ支持消息应答机制。
             当消费者接收到消息并完成任务后会往RabbitMQ服务器发送一条确认的命令，然后RabbitMQ才会将消息删除。*/
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        });
        return simpleMessageListenerContainer;
    }

    @Bean("myRabbitListenerContainerFactory")
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        //SimpleRabbitListenerContainerFactory发现消息中有content_type有text就会默认将其转换成string类型的
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setMessageConverter(new MessageConverter() {
            @Override
            public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException {
                return null;
            }

            @Override
            public Object fromMessage(Message message) throws MessageConversionException {
                return new String(message.getBody());
            }
        });
        return factory;
    }

    /**
     * 定义rabbit template用于数据的接收和发送
     *
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        /**若使用confirm-callback或return-callback，
         * 必须要配置publisherConfirms或publisherReturns为true
         * 每个rabbitTemplate只能有一个confirm-callback和return-callback
         */
        template.setConfirmCallback(msgSendConfirmCallBack());
        /**
         * 使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true，
         * 可针对每次请求的消息去确定’mandatory’的boolean值，
         * 只能在提供’return -callback’时使用，与mandatory互斥
         */
        template.setMandatory(true);
        template.setReturnCallback(msgSendReturnCallBack());
        return template;
    }

    /**
     * 消息确认机制
     * Confirms给客户端一种轻量级的方式，能够跟踪哪些消息被broker处理，
     * 哪些可能因为broker宕掉或者网络失败的情况而重新发布。
     * 确认并且保证消息被送达，提供了两种方式：发布确认和事务。(两者不可同时使用)
     * 在channel为事务时，不可引入确认模式；同样channel为确认模式下，不可使用事务。
     *
     * @return
     */
    @Bean
    public MsgSendConfirmCallBack msgSendConfirmCallBack() {
        return new MsgSendConfirmCallBack();
    }

    @Bean
    public MsgSendReturnCallBack msgSendReturnCallBack() {
        return new MsgSendReturnCallBack();
    }

}
